package com.atguigu.flink.watermark;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2023/2/28
 *
 *  指定在哪里产生watermark.
 *          在源头
 *          在中间转换的位置。
 *
 *   assignTimestampsAndWatermarks(WatermarkStrategy 水印的制作策略)
 *
 *      WatermarkStrategy：
 *          ①告诉策略，数据的哪个属性是 时间属性
 *          ②选择策略，对时间属性运算，生成水印
 *                  WatermarkStrategy.forMonotonousTimestamps(): 基于非乱序的场景生成水印。默认水印 = 数据的时间属性 - 1ms
 *                   WatermarkStrategy.forBoundedOutOfOrderness(延迟 n): 基于乱序的场景生成水印。 默认水印 = 数据的时间属性 - n ms
 *
 *                          以上两个策略用的都是BoundedOutOfOrdernessWatermarks的构造器。
 *
 *                    WatermarkStrategy.noWatermarks(): 不生成
 *                     WatermarkStrategy.forGenerator(): 自定义策略
 *
 *     -----------------------------------
 *      如果有两个同名的类，JVM虚拟机在加载时，基于双亲委派模型会优先加载用户工程目录下的类。
 *
 *      如何去调试Flink的源码?
 *              在自己的工程下创建和源码一样的包，复制源码到自己的项目下，优先加载。
 *
 *
 */
public class Demo1_Watermark
{
    public static void main(String[] args) {

       Configuration conf = new Configuration();
       conf.setInteger("rest.port", 3333);
       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

       env.disableOperatorChaining();
       //设置水印周期发送的间隔
       env.getConfig().setAutoWatermarkInterval(2000);


        //构造水印的制作策略
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            //选择策略
            .<WaterSensor>forMonotonousTimestamps()
            //告诉策略，数据的哪个属性是 时间属性
            .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>()
            {

                @Override
                public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                    //返回毫秒时间戳
                    return element.getTs();
                }
            });

        //刚上手，并行度必须是1
        env.setParallelism(1);


        env
                   .socketTextStream("hadoop103", 8888)
                   .map(new WaterSensorMapFunction())
                   //为数据分配 时间戳的水印。 水印产生后，会向下游流动
                   .assignTimestampsAndWatermarks(watermarkStrategy)
                   .process(new ProcessFunction<WaterSensor, WaterSensor>()
                   {
                       @Override
                       public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {

                           //获取当前算子的水印
                           TimerService timerService = ctx.timerService();
                           /*
                                currentWatermark: 当前数据到达这个process算子时，这个算子的时间是几点
                                    watermark:  水印，代表flink中的时间

                                    更新前WM
                            */
                           System.out.println(value+"到达时,此时process的WM:"+timerService.currentWatermark());

                       }
                   })
                   .map(x -> x)
                   .print();


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
